package com.nami.gate.router.middleware.mq.consumer.listener;

import cn.hutool.core.collection.CollUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;

public abstract class AbstractConsumer implements MessageListenerConcurrently {

    private static final Logger log = LoggerFactory.getLogger(AbstractConsumer.class);

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                    ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(list)) { // CollUtil.isEmpty()
            log.error("[down-abstract-consumer] 当前topic：{} 出现错误了 数据没有传过来", getTopicName());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        for (MessageExt messageExt : list) {
            onMessage(messageExt.getBody());
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public abstract String getTopicName();

    public abstract void onMessage(byte[] body);

}
